package com.swak.job.queue.rabbitmq;

import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.function.Function;

import com.swak.job.JobTask;
import com.swak.job.queue.QueueService;
import com.swak.rabbit.EventBusService;
import com.swak.rabbit.RabbitMQTemplate;
import com.swak.rabbit.TemplateConsumer;
import com.swak.utils.Lists;
import com.swak.utils.Maps;

/**
 * 队列服务： 消费失败直接丢弃
 * 
 * @author DELL
 */
public class RabbitmqQueueService extends EventBusService implements QueueService {

	Map<Function<JobTask, Object>, List<TemplateConsumer>> consumers = Maps.newConcurrentMap();

	public RabbitmqQueueService(RabbitMQTemplate template) {
		super(template);
	}

	@Override
	public void createQueue(String queue) {
		this.template.exchangeFanoutBindQueue(queue, queue, null);
	}

	@Override
	public void publishQueue(String queue, JobTask task) {
		this.post(queue, queue, task);
	}

	@Override
	public CompletableFuture<Void> publishQueueAsync(String queue, JobTask task) {
		return this.submit(queue, queue, task);
	}

	@Override
	public void consumeQueue(String queue, Function<JobTask, Object> func) {
		TemplateConsumer consumer = this.template.basicConsume(queue, 1, (msg) -> {
			JobTask task = msg.payload2Object();
			return func.apply(task);
		});
		List<TemplateConsumer> lists = consumers.computeIfAbsent(func, (key) -> {
			return Lists.newArrayList();
		});
		lists.add(consumer);
	}

	@Override
	public void unconsumeQueue(String queue, Function<JobTask, Object> func) {
		List<TemplateConsumer> lists = consumers.computeIfAbsent(func, (key) -> {
			return Lists.newArrayList();
		});
		lists.forEach(consumer -> {
			consumer.cancel();
		});
		lists.clear();
	}
}
